package com.surfilter.massdata.spark.output.ipcheck;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Types;
import java.util.Date;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import com.act.sparkanalyz.output.impl.HiveOutput;
import com.surfilter.massdata.spark.util.JdbcUtilNew;
import com.surfilter.massdata.spark.util.ipcheck.IpCheckUtil;
import com.surfilter.massdata.spark.util.ipcheck.JavaForeachPartitionDFFunc;

import scala.collection.Iterator;

/** 
* @author  hutao
* @date 创建时间：2017年6月9日 上午10:54:39 
* @Description：将数据入库oracle的br7006(基准来源表)
* @version 1.0 
* 
**/
public class IpCheckToOracleBr7006 extends HiveOutput {

	private static final long serialVersionUID = 1L;
	private static final Log log = LogFactory.getLog(IpCheckToOracleBr7006.class);
	private String tableName;
	@Override
	public void write(DataFrame result, Map<String, String> commandMap) {
		
		log.info("===========IpCheckToOracleBr7006 begin============");
		try{
//			IpCheckUtil.truncateTable(tableName);//清空表数据
			System.err.println("br7006 result>>>>>>>>>>>>>>>>>>>>>>>>"+result.count());
			writeToTable(result,tableName);
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
		
		log.info("===========IpCheckToOracleBr7006 end============");
	}
	

	public static void writeToTable(DataFrame df,String tableName) throws Exception {

		final String br7006InsertSql = "INSERT INTO "+tableName+"(ID,BBDW_ID,LYDW,QSIP,ZZIP,BZ,SOURCE_IDC_ID,REPORT_IDC_ID,REPORT_TYPE,LAST_MODIFY_TIME)" +
				" VALUES(BR7006_SE.NEXTVAL, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
		df.foreachPartition(new JavaForeachPartitionDFFunc() {
			private static final long serialVersionUID = 1L;

			@Override
			public void call(Iterator<Row> it) {
				// TODO Auto-generated method stub
				Connection conn = null;
				PreparedStatement ps7006 = null;
				try {
					conn = JdbcUtilNew.getConnection();
					conn.setAutoCommit(false);
					int count = 0;
					int batchSize = 5000;
					ps7006 = conn.prepareStatement(br7006InsertSql);
					while (it.hasNext()) {
						Row row = it.next();
						ps7006.setLong(1, Long.parseLong(row.getAs("bbdw_id").toString()));
						ps7006.setLong(2, Long.parseLong(row.getAs("lyid").toString()));
						ps7006.setLong(3, Long.parseLong(row.getAs("qsip").toString()));
						ps7006.setLong(4, Long.parseLong(row.getAs("zzip").toString()));
						if(row.getAs("bz") == null){
							ps7006.setNull(5, Types.VARCHAR);
						}else{
							ps7006.setString(5, row.getAs("bz").toString());
						}
						if(row.getAs("sourceIdcId")  == null||((String)row.getAs("sourceIdcId")).equals("N/A")){
							ps7006.setNull(6, Types.VARCHAR);
						}else{
							ps7006.setString(6, row.getAs("sourceIdcId").toString());
						}
						if(row.getAs("reportIdcId")== null||((String)row.getAs("reportIdcId")).equals("N/A")){
							ps7006.setNull(7, Types.VARCHAR);
						}else{
							ps7006.setString(7, row.getAs("reportIdcId").toString());
						}
						
						if(row.getAs("reportType")== null){
							ps7006.setNull(8, Types.NUMERIC);
						}else{
							ps7006.setInt(8, Integer.parseInt(row.getAs("reportType").toString()));
						}
						if(row.getAs("lastModifyTime")==null){
							ps7006.setDate(9, new java.sql.Date(new Date().getTime()));
						}else{
							ps7006.setDate(9, new java.sql.Date((Long)row.getAs("lastModifyTime")));
						}
						
						ps7006.addBatch();
						if(++count % batchSize==0){
							ps7006.executeBatch();
							conn.commit();
							ps7006.clearBatch();
							count = 0;
						}
						
					}//end while
					ps7006.executeBatch();
					conn.commit();
					ps7006.clearBatch();

				} catch (Exception e) {
					e.printStackTrace();
				} finally {
					if (null != ps7006) {
						try {
							ps7006.close();
						} catch (SQLException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
					}
					if (conn != null) {
						try {
							conn.close();
						} catch (SQLException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
					}
				}

			}
		});

	}
}
